package com.lefeng.cockpit.resource.service.impl;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.didiglobal.logi.security.common.PagingData;
import com.didiglobal.logi.security.common.Result;
import com.didiglobal.logi.security.util.CopyBeanUtil;
import com.google.common.base.Joiner;
import com.google.common.io.Files;
import com.google.protobuf.ServiceException;
import com.lefeng.cockpit.common.bean.dto.resource.ResourceDTO;
import com.lefeng.cockpit.common.bean.po.resource.ResourcePO;
import com.lefeng.cockpit.common.config.UserLocal;
import com.lefeng.cockpit.common.enums.resource.ResourceType;
import com.lefeng.cockpit.common.enums.resource.Status;
import com.lefeng.cockpit.resource.dao.ResourceDao;
import com.lefeng.cockpit.resource.service.ResourcesService;
import com.lefeng.cockpit.resource.storage.StorageOperate;
import com.lefeng.cockpit.resource.utils.FileUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanMap;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.MessageFormat;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class ResourcesServiceImpl implements ResourcesService {


    private static final String FORMAT_SS = "%s%s";
    private static final String FORMAT_S_S = "%s/%s";
    private static final String FOLDER_SEPARATOR = "/";
    private static final int MAX_FILE_SIZE = 1024 * 1024 * 1024;
    private static final String ALIAS = "alias";
    private static final String CONTENT = "content";

    @Resource
    private ResourceDao resourceDao;

    @Resource
    private StorageOperate storageOperate;


    @Override
    public Result<Object> createDirectory(ResourceDTO dto) throws ServiceException {
        Result<Object> result = new Result<>();

        result = checkResourceUploadStartupState();
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }
        if (FileUtils.directoryTraversal(dto.getAlias())) {
            putMsg(result, Status.VERIFY_PARAMETER_NAME_FAILED);
            return result;
        }

        if (checkDescriptionLength(dto.getDescription())) {
            putMsg(result, Status.DESCRIPTION_TOO_LONG_ERROR);
            return result;
        }

        String fullName = getFullName(dto.getCurrentDir(), dto.getAlias());
        result = verifyResource(dto.getType(), fullName, dto.getPid());

        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }

        if (checkResourceExists(fullName, dto.getType().ordinal())) {
            log.error("resourcePO directory {} has exist, can't recreate", fullName);
            putMsg(result, Status.RESOURCE_EXIST);
            return result;
        }

        ResourcePO entity = CopyBeanUtil.copy(dto, ResourcePO.class);
        entity.setType(dto.getType().getCode());
        entity.setDirectory(true);
        entity.setUserName(UserLocal.getUser());


        try {
            resourceDao.insert(entity);
            putMsg(result, Status.SUCCESS);
            Map<String, Object> resultMap = new HashMap<>();
            for (Map.Entry<Object, Object> entry : new BeanMap(entity).entrySet()) {
                if (!"class".equalsIgnoreCase(entry.getKey().toString())) {
                    resultMap.put(entry.getKey().toString(), entry.getValue());
                }
            }
            result.setData(resultMap);
        } catch (DuplicateKeyException e) {
            log.error("resource directory {} has exist, can't recreate", fullName);
            putMsg(result, Status.RESOURCE_EXIST);
            return result;
        } catch (Exception e) {
            log.error("resource already exists, can't recreate ", e);
            throw new RuntimeException("resource already exists, can't recreate");
        }
        // create directory in storage
        createDirectory(fullName, dto.getType(), result);
        return result;
    }

    @Override
    public Result<Object> createResource(ResourceDTO dto) throws ServiceException {
        Result<Object> result = new Result<>();

        result = checkResourceUploadStartupState();
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }

        result = verifyPid(dto.getPid());
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }
        if (checkDescriptionLength(dto.getDescription())) {
            putMsg(result, Status.DESCRIPTION_TOO_LONG_ERROR);
            return result;
        }

        result = verifyFile(dto.getAlias(), dto.getType(), dto.getFile());
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }

        // check resource name exists
        String fullName = getFullName(dto.getCurrentDir(), dto.getAlias());
        if (checkResourceExists(fullName, dto.getType().ordinal())) {
            log.error("resource {} has exist, can't recreate", escapeNRT(dto.getAlias()));
            putMsg(result, Status.RESOURCE_EXIST);
            return result;
        }
        if (fullName.length() > 128) {
            log.error("resource {}'s full name {}' is longer than the max length {}", escapeNRT(dto.getAlias()),
                    fullName, 128);
            putMsg(result, Status.RESOURCE_FULL_NAME_TOO_LONG_ERROR);
            return result;
        }

        ResourcePO resource = CopyBeanUtil.copy(dto, ResourcePO.class);
        resource.setType(dto.getType().getCode());
        resource.setDirectory(false);
        resource.setUserName(UserLocal.getUser());
        resource.setSize(dto.getFile().getSize());
        resource.setFileName(dto.getFile().getOriginalFilename());
        resource.setFullName(fullName);

        try {
            resourceDao.insert(resource);
            updateParentResourceSize(resource, resource.getSize());
            putMsg(result, Status.SUCCESS);
            Map<String, Object> resultMap = new HashMap<>();
            for (Map.Entry<Object, Object> entry : new BeanMap(resource).entrySet()) {
                if (!"class".equalsIgnoreCase(entry.getKey().toString())) {
                    resultMap.put(entry.getKey().toString(), entry.getValue());
                }
            }
            result.setData(resultMap);
        } catch (Exception e) {
            log.error("resource already exists, can't recreate ", e);
            throw new ServiceException("resource already exists, can't recreate");
        }

        // fail upload
        if (!upload(fullName, dto.getFile(), dto.getType())) {
            log.error("upload resource: {} file: {} failed.", escapeNRT(dto.getAlias()),
                    escapeNRT(dto.getFile().getOriginalFilename()));
            putMsg(result, Status.STORE_OPERATE_CREATE_ERROR);
            throw new ServiceException(
                    String.format("upload resource: %s file: %s failed.", dto.getAlias(), dto.getFile().getOriginalFilename()));
        }
        return result;
    }

    @Override
    public Result onlineCreateResource(ResourceDTO dto) throws ServiceException {
        Result<Object> result = new Result<>();


        result = checkResourceUploadStartupState();
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }
        if (FileUtils.directoryTraversal(dto.getFileName())) {
            putMsg(result, Status.VERIFY_PARAMETER_NAME_FAILED);
            return result;
        }
        if (checkDescriptionLength(dto.getDescription())) {
            putMsg(result, Status.DESCRIPTION_TOO_LONG_ERROR);
            return result;
        }

        // check file suffix
        String nameSuffix = dto.getFileSuffix().trim();
        String resourceViewSuffixes = FileUtils.getResourceViewSuffixes();
        if (org.apache.commons.lang3.StringUtils.isNotEmpty(resourceViewSuffixes)) {
            List<String> strList = Arrays.asList(resourceViewSuffixes.split(","));
            if (!strList.contains(nameSuffix)) {
                log.error("resource suffix {} not support create", nameSuffix);
                putMsg(result, Status.RESOURCE_SUFFIX_NOT_SUPPORT_VIEW);
                return result;
            }
        }

        String name = dto.getFileName().trim() + "." + nameSuffix;
        String fullName = getFullName(dto.getCurrentDir(), name);
        result = verifyResource(dto.getType(), fullName, dto.getPid());
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }

        // save data
        ResourcePO resource = CopyBeanUtil.copy(dto, ResourcePO.class);
        resource.setAlias(name);
        resource.setDirectory(false);
        resource.setUserName(UserLocal.getUser());
        resource.setSize(dto.getContent().getBytes().length);
        resource.setFileName(name);
        resource.setFullName(fullName);
        resource.setType(dto.getType().getCode());

        resourceDao.insert(resource);
        updateParentResourceSize(resource, resource.getSize());

        putMsg(result, Status.SUCCESS);
        Map<String, Object> resultMap = new HashMap<>();
        for (Map.Entry<Object, Object> entry : new BeanMap(resource).entrySet()) {
            if (!"class".equalsIgnoreCase(entry.getKey().toString())) {
                resultMap.put(entry.getKey().toString(), entry.getValue());
            }
        }
        result.setData(resultMap);

        result = uploadContentToStorage(fullName, dto.getContent());
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            throw new ServiceException(result.getMessage());
        }
        return result;
    }

    @Override
    public Result updateResourceContent(ResourceDTO dto) throws ServiceException {
        Result<Object> result = checkResourceUploadStartupState();
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }

        ResourcePO resource = resourceDao.selectById(dto.getResourceId());
        if (resource == null) {
            log.error("read file not exist,  resource id {}", dto.getResourceId());
            putMsg(result, Status.RESOURCE_NOT_EXIST);
            return result;
        }

        // check can edit by file suffix
        String nameSuffix = Files.getFileExtension(resource.getAlias());
        String resourceViewSuffixes = FileUtils.getResourceViewSuffixes();
        if (org.apache.commons.lang3.StringUtils.isNotEmpty(resourceViewSuffixes)) {
            List<String> strList = Arrays.asList(resourceViewSuffixes.split(","));
            if (!strList.contains(nameSuffix)) {
                log.error("resource suffix {} not support updateProcessInstance,  resource id {}", nameSuffix,
                        dto.getResourceId());
                putMsg(result, Status.RESOURCE_SUFFIX_NOT_SUPPORT_VIEW);
                return result;
            }
        }

        long originFileSize = resource.getSize();
        resource.setSize(dto.getContent().getBytes().length);
        resource.setUpdateTime(new Date());
        resourceDao.updateById(resource);

        result = uploadContentToStorage(resource.getFullName(), dto.getContent());
        updateParentResourceSize(resource, resource.getSize() - originFileSize);

        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            throw new ServiceException(result.getMessage());
        }
        return result;
    }

    @Override
    public Result<Object> delete(int resourceId) throws IOException {
        // get resource by id
        Result<Object> resultCheck = new Result<>();
        ResourcePO resource = resourceDao.selectById(resourceId);
        if (resource == null) {
            putMsg(resultCheck, Status.RESOURCE_NOT_EXIST);
            return resultCheck;
        }

        Result<Object> result = checkResourceUploadStartupState();
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }

        // get all children of the resource
        List<Integer> allChildren = listAllChildren(resource, true);
        Integer[] needDeleteResourceIdArray = allChildren.toArray(new Integer[allChildren.size()]);

        // get hdfs file by type
        String storageFilename = storageOperate.getFileName(resource.getType() == 0 ? ResourceType.FILE : ResourceType.UDF, resource.getFullName());
        // delete data in database
        resourceDao.selectBatchIds(Arrays.asList(needDeleteResourceIdArray)).forEach(item -> {
            updateParentResourceSize(item, item.getSize() * -1);
        });
        resourceDao.deleteIds(needDeleteResourceIdArray);

        // delete file on hdfs

        // delete file on storage
        storageOperate.delete(storageFilename, true);
        putMsg(result, Status.SUCCESS);

        return result;
    }

    @Override
    public PagingData<ResourcePO> queryResourceListPaging(ResourceDTO dto) throws ServiceException {
        if (dto.getPid() != -1) {
            ResourcePO directory = resourceDao.selectById(dto.getPid());
            if (directory == null) {
                throw new ServiceException(Status.RESOURCE_NOT_EXIST.getMsg());
            }
        }

        IPage<ResourcePO> dataSourcePOs = resourceDao.queryResourcePaging(dto);
        List<ResourcePO> dataSourceVOS = dataSourcePOs.getRecords();
        return new PagingData<>(dataSourceVOS, dataSourcePOs);
    }

    @Override
    public Result readResource(int resourceId, int skipLineNum, int limit) {
        Result<Object> result = checkResourceUploadStartupState();
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }
        // get resource by id
        ResourcePO resource = resourceDao.selectById(resourceId);
        if (resource == null) {
            putMsg(result, Status.RESOURCE_NOT_EXIST);
            return result;
        }

        // check preview or not by file suffix
        String nameSuffix = Files.getFileExtension(resource.getAlias());
        String resourceViewSuffixes = FileUtils.getResourceViewSuffixes();
        if (org.apache.commons.lang3.StringUtils.isNotEmpty(resourceViewSuffixes)) {
            List<String> strList = Arrays.asList(resourceViewSuffixes.split(","));
            if (!strList.contains(nameSuffix)) {
                log.error("resource suffix {} not support view,  resource id {}", nameSuffix, resourceId);
                putMsg(result, Status.RESOURCE_SUFFIX_NOT_SUPPORT_VIEW);
                return result;
            }
        }
        // source path
        String resourceFileName = storageOperate.getResourceFileName(resource.getFullName());
        log.info("resource  path is {}", resourceFileName);
        try {
            if (storageOperate.exists(resourceFileName)) {
                List<String> content = storageOperate.vimFile(resourceFileName, skipLineNum, limit);

                putMsg(result, Status.SUCCESS);
                Map<String, Object> map = new HashMap<>();
                map.put(ALIAS, resource.getAlias());
                map.put(CONTENT, String.join("\n", content));
                result.setData(map);
            } else {
                log.error("read file {} not exist in storage", resourceFileName);
                putMsg(result, Status.RESOURCE_FILE_NOT_EXIST, resourceFileName);
            }

        } catch (Exception e) {
            log.error("Resource {} read failed", resourceFileName, e);
            putMsg(result, Status.HDFS_OPERATION_ERROR);
        }

        return result;
    }

    @Override
    public Result getTree() {
        // 查询所有文件和目录
        List<ResourcePO> fileEntities = resourceDao.selectList();

        // 按照 parentId 分组
        Map<Integer, List<ResourcePO>> groupByParentId = fileEntities.stream()
                .collect(Collectors.groupingBy(ResourcePO::getPid));

        // 构造树形结构
        return Result.buildSucc(buildTree(groupByParentId, -1));
    }


//    ========================================= private ==========================================

    /**
     * build resource tree
     *
     * @param groupByParentId groupByParentId
     * @param parentId        parentId
     * @return ..
     */
    private List<Map<String, Object>> buildTree(Map<Integer, List<ResourcePO>> groupByParentId, Integer parentId) {
        List<Map<String, Object>> result = new ArrayList<>();

        // 获取当前父节点的子节点列表
        List<ResourcePO> children = groupByParentId.get(parentId);
        if (children != null) {
            for (ResourcePO entity : children) {
                // 构建节点
                Map<String, Object> node = new HashMap<>();
                node.put("title", entity.getFileName());
                node.put("key", entity.getId().toString());

                // 如果是目录，递归构建子节点
                if (entity.isDirectory()) {
                    List<Map<String, Object>> childNodes = buildTree(groupByParentId, entity.getId());
                    node.put("children", childNodes);
                } else {
                    // 是叶子节点
                    node.put("isLeaf", true);
                }

                result.add(node);
            }
        }

        return result;
    }

    /**
     * list all children id
     *
     * @param resource    resource
     * @param containSelf whether add self to children list
     * @return all children id
     */
    List<Integer> listAllChildren(ResourcePO resource, boolean containSelf) {
        List<Integer> childList = new ArrayList<>();
        if (resource.getId() != null && containSelf) {
            childList.add(resource.getId());
        }

        if (resource.isDirectory()) {
            listAllChildren(resource.getId(), childList);
        }
        return childList;
    }

    /**
     * list all children id
     *
     * @param resourceId resource id
     * @param childList  child list
     */
    void listAllChildren(int resourceId, List<Integer> childList) {
        List<Integer> children = resourceDao.listChildren(resourceId);
        for (int childId : children) {
            childList.add(childId);
            listAllChildren(childId, childList);
        }
    }

    private String getFullName(String currentDir, String name) {
        return currentDir.equals(FOLDER_SEPARATOR) ? String.format(FORMAT_SS, currentDir, name)
                : String.format(FORMAT_S_S, currentDir, name);
    }

    private Result<Object> checkResourceUploadStartupState() {
        Result<Object> result = new Result<>();
        putMsg(result, Status.SUCCESS);
        // if resource upload startup
//        if (!PropertyUtils.getResUploadStartupState()) {
//            log.error("resource upload startup state: {}", PropertyUtils.getResUploadStartupState());
//            putMsg(result, Status.STORAGE_NOT_STARTUP);
//            return result;
//        }
        return result;
    }

    private void putMsg(Result result, Status status, Object... statusParams) {
        result.setCode(status.getCode());
        if (statusParams != null && statusParams.length > 0) {
            result.setMessage(MessageFormat.format(status.getMsg(), statusParams));
        } else {
            result.setMessage(status.getMsg());
        }
    }

    private boolean checkDescriptionLength(String description) {
        return description != null && description.codePointCount(0, description.length()) > 255;
    }

    /**
     * check resource is exists
     *
     * @param fullName fullName
     * @param type     type
     * @return true if resource exists
     */
    private boolean checkResourceExists(String fullName, int type) {
        Boolean existResource = resourceDao.existResource(fullName, type);
        return Boolean.TRUE.equals(existResource);
    }

    /**
     * verify resource by name and type
     *
     * @param fullName resource full name
     * @param type     resource type
     * @return true if the resource name not exists, otherwise return false
     */
    private Result<Object> verifyResourceName(String fullName, ResourceType type) {
        Result<Object> result = new Result<>();

        putMsg(result, Status.SUCCESS);
        if (checkResourceExists(fullName, type.ordinal())) {
            log.error("resource type:{} name:{} has exist, can't create again.", type,
                    escapeNRT(fullName));
            putMsg(result, Status.RESOURCE_EXIST);
        } else {

            try {
                String filename = storageOperate.getFileName(type, fullName);
                if (storageOperate.exists(filename)) {
                    putMsg(result, Status.RESOURCE_FILE_EXIST, filename);
                }

            } catch (Exception e) {
                log.error("verify resource failed  and the reason is {}", e.getMessage());
                putMsg(result, Status.STORE_OPERATE_CREATE_ERROR);
            }

        }

        return result;
    }

    private static String escapeNRT(String str) {
        // Logging should not be vulnerable to injection attacks: Replace pattern-breaking characters
        if (!StringUtils.isEmpty(str)) {
            return str.replaceAll("[\n|\r|\t]", "_");
        }
        return null;
    }


    private Result<Object> verifyResource(ResourceType type, String fullName, int pid) {
        Result<Object> result = verifyResourceName(fullName, type);
        if (!result.getCode().equals(Status.SUCCESS.getCode())) {
            return result;
        }
        return verifyPid(pid);
    }

    private Result<Object> verifyPid(int pid) {
        Result<Object> result = new Result<>();
        putMsg(result, Status.SUCCESS);
        if (pid != -1) {
            ResourcePO parentResource = resourceDao.selectById(pid);
            if (parentResource == null) {
                putMsg(result, Status.PARENT_RESOURCE_NOT_EXIST);
                return result;
            }
        }
        return result;
    }

    /**
     * create directory
     * xxx The steps to verify resources are cumbersome and can be optimized
     *
     * @param fullName full name
     * @param type     resource type
     * @param result   Result
     */
    private void createDirectory(String fullName, ResourceType type, Result<Object> result) throws ServiceException {

        String directoryName = storageOperate.getFileName(type, fullName);
        String resourceRootPath = storageOperate.getDir(type);
        try {
            if (!storageOperate.exists(resourceRootPath)) {
                storageOperate.createTenantDirIfNotExists();
            }

            if (!storageOperate.mkdir(directoryName)) {
                log.error("create resource directory {}  failed", directoryName);
                putMsg(result, Status.STORE_OPERATE_CREATE_ERROR);
                throw new ServiceException(String.format("create resource directory: %s failed.", directoryName));
            }
        } catch (Exception e) {
            log.error("create resource directory {}  failed", directoryName);
            putMsg(result, Status.STORE_OPERATE_CREATE_ERROR);
            throw new ServiceException(String.format("create resource directory: %s failed.", directoryName));
        }
    }

    /**
     * update the folder's size of the resource
     *
     * @param resource the current resource
     * @param size     size
     */
    private void updateParentResourceSize(ResourcePO resource, long size) {
        if (resource.getSize() > 0) {
            String[] splits = resource.getFullName().split("/");
            for (int i = 1; i < splits.length; i++) {
                String parentFullName = Joiner.on("/").join(Arrays.copyOfRange(splits, 0, i));
                if (org.apache.commons.lang3.StringUtils.isNotBlank(parentFullName)) {
                    List<ResourcePO> resources =
                            resourceDao.queryResource(parentFullName, resource.getType());
                    if (CollectionUtils.isNotEmpty(resources)) {
                        ResourcePO parentResource = resources.get(0);
                        if (parentResource.getSize() + size >= 0) {
                            parentResource.setSize(parentResource.getSize() + size);
                        } else {
                            parentResource.setSize(0L);
                        }
                        resourceDao.updateById(parentResource);
                    }
                }
            }
        }
    }

    /**
     * upload file to hdfs
     *
     * @param fullName full name
     * @param file     file
     */
    private boolean upload(String fullName, MultipartFile file, ResourceType type) {
        // save to local
        String fileSuffix = Files.getFileExtension(file.getOriginalFilename());
        String nameSuffix = Files.getFileExtension(fullName);

        // determine file suffix
        if (!fileSuffix.equalsIgnoreCase(nameSuffix)) {
            return false;
        }
        // random file name
        String localFilename = FileUtils.getUploadFilename(UUID.randomUUID().toString());

        // save file to hdfs, and delete original file
        String fileName = storageOperate.getFileName(type, fullName);
        String resourcePath = storageOperate.getDir(type);
        try {
            // if tenant dir not exists
            if (!storageOperate.exists(resourcePath)) {
                storageOperate.createTenantDirIfNotExists();
            }
            FileUtils.copyInputStreamToFile(file, localFilename);
            storageOperate.upload(localFilename, fileName, true, true);
        } catch (Exception e) {
            FileUtils.deleteFile(localFilename);
            log.error(e.getMessage(), e);
            return false;
        }
        return true;
    }

    private Result<Object> verifyFile(String name, ResourceType type, MultipartFile file) {
        Result<Object> result = new Result<>();
        putMsg(result, Status.SUCCESS);

        if (FileUtils.directoryTraversal(name)) {
            log.error("file alias name {} verify failed", name);
            putMsg(result, Status.VERIFY_PARAMETER_NAME_FAILED);
            return result;
        }

        if (file != null && FileUtils.directoryTraversal(Objects.requireNonNull(file.getOriginalFilename()))) {
            log.error("file original name {} verify failed", file.getOriginalFilename());
            putMsg(result, Status.VERIFY_PARAMETER_NAME_FAILED);
            return result;
        }

        if (file != null) {
            // file is empty
            if (file.isEmpty()) {
                log.error("file is empty: {}", escapeNRT(file.getOriginalFilename()));
                putMsg(result, Status.RESOURCE_FILE_IS_EMPTY);
                return result;
            }

            // file suffix
            String fileSuffix = Files.getFileExtension(file.getOriginalFilename());
            String nameSuffix = Files.getFileExtension(name);

            // determine file suffix
            if (!fileSuffix.equalsIgnoreCase(nameSuffix)) {
                // rename file suffix and original suffix must be consistent
                log.error("rename file suffix and original suffix must be consistent: {}",
                        escapeNRT(file.getOriginalFilename()));
                putMsg(result, Status.RESOURCE_SUFFIX_FORBID_CHANGE);
                return result;
            }

            if (file.getSize() > MAX_FILE_SIZE) {
                log.error("file size is too large: {}", escapeNRT(file.getOriginalFilename()));
                putMsg(result, Status.RESOURCE_SIZE_EXCEED_LIMIT);
                return result;
            }
        }
        return result;
    }

    /**
     * @param resourceName resource name
     * @param content      content
     * @return result
     */
    private Result<Object> uploadContentToStorage(String resourceName,
                                                  String content) {
        Result<Object> result = new Result<>();
        String localFilename = "";
        String storageFileName = "";
        try {
            localFilename = FileUtils.getUploadFilename(UUID.randomUUID().toString());

            if (!FileUtils.writeContent2File(content, localFilename)) {
                // write file fail
                log.error("file {} fail, content is {}", localFilename, escapeNRT(content));
                putMsg(result, Status.RESOURCE_NOT_EXIST);
                return result;
            }

            // get resource file path
            storageFileName = storageOperate.getResourceFileName(resourceName);
            String resourcePath = storageOperate.getResDir();
            log.info("resource  path is {}, resource dir is {}", storageFileName, resourcePath);

            if (!storageOperate.exists(resourcePath)) {
                // create if tenant dir not exists
                storageOperate.createTenantDirIfNotExists();
            }
            if (storageOperate.exists(storageFileName)) {
                storageOperate.delete(storageFileName, false);
            }

            storageOperate.upload(localFilename, storageFileName, true, true);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            result.setCode(Status.HDFS_OPERATION_ERROR.getCode());
            result.setMessage(String.format("copy %s to hdfs %s fail", localFilename, storageFileName));
            return result;
        }
        putMsg(result, Status.SUCCESS);
        return result;
    }
}
